package cn.lsh.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.*;
import java.net.URI;

public class FileSystemUtil {

	private static FileSystem fs;

	// public static final String HDFS_PATH = "hdfs://192.168.137.229:9000";
	public static final String HDFS_PATH = "hdfs://node00:8020";

	/**
	 * 解决无权限访问,设置远程hadoop的linux用户名称
	 */
	public static final String USER = "root";

	static {
		fs = getFileSystem();
	}

	public static FileSystem getFileSystem() {
		return getFileSystem(HDFS_PATH, USER);
	}

	public static FileSystem getFileSystem(String host, int port) {
		return getFileSystem(host, port, null);
	}

	public static FileSystem getFileSystem(String host, int port, String user) {
		return getFileSystem("hdfs://" + host + ":" + port, user);
	}

	private static FileSystem getFileSystem(String hdfsPath, String user) {
		try {
			URI hdsfUri = URI.create(hdfsPath);
			Configuration cfg = new Configuration();
			//代码修改hdfs-site.xml设置，也可以直接在hadoop安装配置文件中修改
			//dfs.client.block.write.replace-datanode-on-failure.polic 默认在有三个或以上备份时，会尝试更换节点写入datanode，单机版可以关掉
			cfg.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
			//客户端在写失败的时候，是否使用更换策略，默认是true
			cfg.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
			// return FileSystem.get(hdsfUri, cfg, user);
			return FileSystem.get(cfg);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	public static void open(String path) throws IOException {
		FSDataInputStream inputStream = fs.open(new Path(path));
		BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
		while (reader.ready()) {
			System.out.println(reader.readLine());
		}
		//移动读取文件指针到指定位置
		inputStream.seek(0);
		System.out.println("-----从头开始再读一遍----------");
		IOUtils.copyBytes(inputStream, System.out, 1024, false);
		IOUtils.closeStream(inputStream);
	}

	/**
	 * 上传本地文件到hdfs，如果hdfs没有指定目录，该方法会自动创建
	 *
	 * @param filePath
	 * @param targetPath
	 * @throws IOException
	 */
	public static void upload(String filePath, String targetPath) throws IOException {
		InputStream inputStream = new BufferedInputStream(new FileInputStream(filePath));
		OutputStream outputStream = fs.create(new Path(targetPath), () -> System.out.println("上传64KB数据"));
		//IOUtils工具类写入输出流到输入流，完毕后关闭流
		IOUtils.copyBytes(inputStream, outputStream, 1024, true);
	}

	/**
	 * 列举所有文件路径，空目录不会列出来
	 *
	 * @throws IOException
	 */
	public static void listFiles() throws IOException {
		RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path("/"), true);
		while (iterator.hasNext()) {
			LocatedFileStatus locatedFileStatus = iterator.next();
			System.out.println(locatedFileStatus.getPath());
		}
	}

	/**
	 * 递归列举所有目录和文件
	 *
	 * @throws IOException
	 */
	public static void listDirAndFile() throws IOException {
		list(new Path("/"));
	}

	private static void list(Path path) throws IOException {
		FileStatus[] fss = fs.listStatus(path);
		if (fss != null && fss.length > 0) {
			for (FileStatus fs : fss) {
				System.out.println(fs.getPath());
				if (fs.isDirectory()) {
					list(fs.getPath());
				}
			}
		}
	}
}
